package rocketmq

import (
	"context"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
)

// Subscriber 订阅数据,自动确认
type Subscriber struct {
	rocketmq.PushConsumer
	chanMessage chan []byte
}

func (this *Subscriber) Close() error {
	return this.PushConsumer.Shutdown()
}

func (this *Subscriber) ReadMessage() ([]byte, error) {
	return <-this.chanMessage, nil
}

func NewSubscriber(cfg *SubscribeConfig) (*Subscriber, error) {
	c, err := rocketmq.NewPushConsumer(
		consumer.WithGroupName(cfg.GroupID),
		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{cfg.Endpoint})),
		consumer.WithNamespace(cfg.InstanceId),
		consumer.WithInstance(cfg.InstanceId),
		consumer.WithCredentials(primitive.Credentials{
			AccessKey: cfg.AccessKey,
			SecretKey: cfg.SecretKey,
		}),
	)
	if err != nil {
		return nil, err
	}
	sub := &Subscriber{
		PushConsumer: c,
		chanMessage:  make(chan []byte),
	}

	if err = c.Subscribe(cfg.Topic, consumer.MessageSelector{Expression: "*"},
		func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
			for _, msg := range msgs {
				if cfg.Debug {
					Log.Writef("[订阅][%s] : %s", cfg.Topic, string(msg.Body))
				}
				sub.chanMessage <- msg.Body
			}
			//返回consumer.Rollback会重发
			return consumer.ConsumeSuccess, nil
		}); err != nil {
		return nil, err
	}

	return sub, c.Start()
}
